package logCollect;

import org.apache.commons.io.FileUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.source.AbstractSource;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.charset.Charset;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

//自定义source实现exec的功能和维护偏移量
public class MySource extends AbstractSource implements EventDrivenSource, Configurable {
    //如果监控一个文件目录下的多个文件
    /**
     * 轮询每个文件     liststatus
     * 1.判断文件是否发生变化(新增文件，修改文件，删除文件)  循环拿到list中每一个元素，执行以下逻辑
     * 2.拿到每个文件的偏移量然后读取相应文件的数据封装event发送
     * 3.获取每一个文件的新的偏移量更新到偏移量文件中
     */
    //以上可以使用FileListen工具类
    /**
     * 1.拿到文件目录
     * 2.利用filelisten找到新增的文件（list1），增加数据的文件(list2),删除的文件（list3）
     * 3.拿到list1，创建偏移量文件，读取数据，更新偏移量
     * 拿到list2，读取数据，更新偏移量
     * 拿到list3，删除偏移量文件
     */

    //成员属性
    //获取原始日志
    String filepath;
    //获取偏移量文件的路径
    String offsetpath;
    //获取编码格式
    String charset;
    //获取睡眠的时间间隔
    Integer interval;

    fileRunner fileRunner;

    //实现获取配置资源的方法
    public void configure(Context context) {
        //获取原始日志
        filepath = context.getString("filepath");
        //获取偏移量文件的路径
        offsetpath = context.getString("offsetpath");
        //获取编码格式
        charset = context.getString("charset", "utf-8");
        //获取睡眠的时间间隔
        interval = context.getInteger("interval", 5);
    }

    //实现业务逻辑的方法
    @Override
    public synchronized void start() {
        //创建线程池
        ExecutorService executor = Executors.newSingleThreadExecutor();
        //获取channel发送器
        ChannelProcessor channelProcessor = getChannelProcessor();
        //创建一个runnanle实现类的对象
        fileRunner = new fileRunner(filepath, offsetpath, charset, interval, channelProcessor);
        //线程池执行任务
        executor.submit(fileRunner);

        super.start();
    }

    //创建一个runnable的实现类
    public static class fileRunner implements Runnable {
        //成员属性
        //获取原始日志
        String filepath;
        //获取偏移量文件的路径
        String offsetpath;
        //获取编码格式
        String charset;
        //获取睡眠的时间间隔
        Integer interval;

        //定义偏移量的成员属性
        long offset = 0L;

        //定义日志文件封装类的属性
        RandomAccessFile raf;

        //定义channel发送器的属性
        ChannelProcessor channelProcessor;

        //定义偏移量文件属性
        File offsetFile;

        //定义标志位
        Boolean flag = true;

        public void setFlag(Boolean flag) {
            this.flag = flag;
        }

        //实现两个方法

        /**
         * 构造器方法：
         * 获取外部类中的属性
         * 获取偏移量文件
         * 判断偏移量文件是否存在，如果不存在则创建
         * 读取偏移量文件的偏移量
         * 打开原始日志文件
         * 将位置挪到偏移量所指向的位置
         */
        public fileRunner(String filepath, String offsetpath, String charset, Integer interval, ChannelProcessor channelProcessor) {
            this.filepath = filepath;
            this.offsetpath = offsetpath;
            this.charset = charset;
            this.interval = interval;
            this.channelProcessor = channelProcessor;

            try {
                //将偏移量文件搞成file
                offsetFile = new File(offsetpath);
                //判断偏移量文件是否存在
                if (!offsetFile.exists()) {
                    offsetFile.createNewFile();
                }
                //读取偏移量文件中的偏移量
                String offsetStr = FileUtils.readFileToString(offsetFile);
                //将偏移量转换成long
                if (null != offsetStr && !"".equals(offsetStr)) {
                    //将字符串的偏移量转换成long类型
                    offset = Long.parseLong(offsetStr);
                }
                //将原始日志文件打开，然后挪动位置
                raf = new RandomAccessFile(filepath, "r");
                raf.seek(offset);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        /**
         * run方法
         * 读取原始日志的数据
         * 判断不为空：
         * 封装成event
         * 发送给channel
         * 获取当前位置信息
         * 更新偏移量到偏移量文件
         * 判断为空：
         * 睡一会
         */
        public void run() {
            while (flag) {
                try {
                    //读取数据
                    String line = raf.readLine();
                    //判断line不为空
                    if (StringUtils.isNotEmpty(line)) {
                        //将line从iso-8859-1转换成utf-8
                        line = new String(line.getBytes("ISO-8859-1"), charset);
                        //将line封装成event
                        Event event = EventBuilder.withBody(line.getBytes());
                        //发送给channel
                        channelProcessor.processEvent(event);
                        //获取当前位置
                        offset = raf.getFilePointer();
                        //更新到偏移量文件中
                        FileUtils.writeStringToFile(offsetFile, offset + "");
                    } else {
                        Thread.sleep(interval);
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }


    //实现清除资源的方法
    @Override
    public synchronized void stop() {
        //关闭资源
//        try {
//            if (!Thread.currentThread().isInterrupted()) {
//                Thread.currentThread().wait(500);
//                fileRunner.setFlag(false);
//            }else{
//                fileRunner.setFlag(false);
//            }
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
        super.stop();
    }
}
